package com.asiainfo.dacp.datastash.broker.flow.listener.service;

import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import com.asiainfo.dacp.datastash.broker.common.constant.ZKConstant;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitTaskStateEnum;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitTask;
import com.asiainfo.dacp.datastash.broker.flow.core.domain.Flow;
import com.asiainfo.dacp.datastash.broker.flow.core.service.FlowStateChangeService;
import com.asiainfo.dacp.datastash.broker.redis.service.RedisFlowService;
import com.asiainfo.dacp.datastash.broker.util.ZKUtil;
import com.asiainfo.dacp.datastash.broker.workerctrl.StashWorkExecutorByEngine;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;

@Service
public class TaskExecProcessor {
	private static Logger LOG = LoggerFactory.getLogger(TaskExecProcessor.class);

	@Autowired
	private RedisFlowService redisService;

	@Autowired
	private FlowStateChangeService stateService;
	
	@Autowired
	private StashWorkExecutorByEngine executor;

	@Async
	protected void handle() {
		String flowTask = redisService.fetchTaskWork();
		if (Strings.isNullOrEmpty(flowTask))
			return;
		LOG.info(String.format("[%s] 开始运行.", flowTask));
		Preconditions.checkArgument(flowTask.split("-").length == 3);
		
		String flowSerNo = flowTask.substring(0, flowTask.lastIndexOf("-"));
		String taskSerNo = flowTask.substring(flowTask.lastIndexOf("-") + 1);
		String unitCode = flowSerNo.substring(0, flowSerNo.indexOf("-"));
		
		String lockPath = String.format(ZKConstant.UNIT_LOCK, unitCode);
		
		begin(lockPath, flowSerNo, taskSerNo);
		
		
		end();
	}
	
	
	public void begin(String lockPath, String flowSerNo, String taskSerNo){
		InterProcessMutex lock = null;
		try {
			lock = ZKUtil.getLock(lockPath);
			lock.acquire();
			
			Flow flow = redisService.getFlow(flowSerNo);
			StashUnitTask task = flow.getTask(taskSerNo);
			task.setTaskState(UnitTaskStateEnum.RUNNING);
			redisService.createOrUpdate(flow);
			stateService.taskRunning(task);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (lock != null)
					lock.release();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	
	public void end(){
		InterProcessMutex lock = null;
		try {
			lock = ZKUtil.getLock(lockPath);
			lock.acquire();
			
			Flow flow = redisService.getFlow(flowSerNo);
			StashUnitTask task = flow.getTask(taskSerNo);
			task.setTaskState(UnitTaskStateEnum.RUNNING);
			redisService.createOrUpdate(flow);
			stateService.taskRunning(task);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (lock != null)
					lock.release();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
}
